package com.example.demo.common.aop;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.rsocket.RSocket;
import lombok.RequiredArgsConstructor;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.util.Logger;
import reactor.util.Loggers;

import java.util.HashMap;
import java.util.Map;

@RequiredArgsConstructor
@Component
@Aspect
public class RsAspect {

    private final static Logger log = Loggers.getLogger(WebAspect.class);
    private final Map<String, RSocketRequester> clients = new HashMap<>();
    private ObjectMapper objectMapper;

    @Around("execution(public * com.example.demo..*.rsocket..*.*(..)) && args(rSocketRequester, clientId, ..)")
    public Object clientRegistration(ProceedingJoinPoint pjp, RSocketRequester rSocketRequester, String clientId) throws Throwable {
        long t = System.currentTimeMillis();
        log.info("=> {} :: args: {}", pjp.getSignature().toShortString(), pjp.getArgs());
        rSocketRequester.rsocketClient()
                .source()
                .flatMap(RSocket::onClose)
                .doFirst(() -> {
                    clients.put(clientId, rSocketRequester);
                    log.info("Client: {} CONNECTED. Number of Clients: {}", clientId, clients.size());
                })
                .doOnError(error -> log.warn("Channel to client {} CLOSED", clientId))
                .doFinally(consumer -> {
                    clients.remove(clientId);
                    log.info("Client {} DISCONNECTED. Number of Clients: {}", clientId, clients.size());
                })
                .subscribe();
        Object retVal = pjp.proceed();
        log.info("<= {} :: Execution Time: {}ms", pjp.getSignature().toShortString(), System.currentTimeMillis() - t);
        return retVal;
    }


    @AfterReturning(pointcut = "execution(public * com.example.demo.modules.demo.rsocket.RsDemoController.rsStream(..))", returning = "stringMono")
    public void streamDislikedMovieEvent(Flux<String> stringMono) {
        stringMono.doOnNext(s -> {
            log.info("输出字符串：" + s);
        });
    }
}
